package org.example.aggres;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * reduce 聚合
 * 获取记录最多的狗
 */
public class DataReduce {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Dog> source = env.fromElements(
                new Dog("aa", "k", 1), new Dog("aa", "k", 1), new Dog("aa", "k", 1), new Dog("aa", "k", 1), new Dog("aa", "k", 1),
                new Dog("bb", "b", 5), new Dog("bb", "b", 5), new Dog("bb", "b", 5),
                new Dog("cc", "b", 6), new Dog("cc", "b", 6),
                new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8), new Dog("dd", "u", 8),
                new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4), new Dog("ee", "u", 4),
                new Dog("ff", "b", 4), new Dog("ff", "b", 4), new Dog("ff", "b", 4), new Dog("ff", "b", 4), new Dog("ff", "b", 4), new Dog("ff", "b", 4)
        );
        source
                /**获取记录数*/
                .map((MapFunction<Dog, Tuple2<String, Integer>>) value ->
                        Tuple2.of(value.getName(), 1))
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                /**聚合*/
                .keyBy(data -> data.f0)
                .reduce((ReduceFunction<Tuple2<String, Integer>>) (value1, value2) ->
                        Tuple2.of(value1.f0, value1.f1 + value2.f1))
                /**获取记录最多的狗*/
                .keyBy(data -> "key")
                .reduce((ReduceFunction<Tuple2<String, Integer>>) (value1, value2) ->
                        value1.f1 > value2.f1 ? value1 : value2)
                .print();
        env.execute();
    }
}